package com.ling.rocketmq.consumer;


import com.ling.rocketmq.model.ConsumerMethodModel;
import com.ling.rocketmq.annotation.Rocketmq;
import org.apache.rocketmq.client.exception.MQClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.util.ClassUtils;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class ConsumerProcessor implements ApplicationContextAware, BeanPostProcessor, SmartInitializingSingleton {

    private final static Logger log = LoggerFactory.getLogger(ConsumerProcessor.class);

    private ApplicationContext applicationContext;

    private final ConcurrentMap<Class<?>, List<ConsumerMethodModel>> typeCache = new ConcurrentHashMap<>();

    private final ConcurrentMap<String, ConsumerMethodModel> methodCache = new ConcurrentHashMap<>();


    /**
     * 所有bean初始化完成之后会调用此方法
     */
    @Override
    public void afterSingletonsInstantiated() {
        ConsumerFactory consumerFactory = applicationContext.getBean(ConsumerFactory.class);
        consumerFactory.setConsumerMethodModelMap(methodCache);
        try {
            consumerFactory.start();
        } catch (MQClientException e) {
            log.warn("消费者启动失败", e);
        }

    }

    /**
     * 在 bean 初始化之后执行
     * 用于寻找添加了 @RocketmqConsumer 注解的方法
     */
    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        Class<?> targetClass = AopUtils.getTargetClass(bean);
        if (Proxy.isProxyClass(targetClass) ||
                ClassUtils.isCglibProxyClass(targetClass)) {
            //代理类不处理
            return bean;
        }
        List<ConsumerMethodModel> listeners = this.typeCache.get(targetClass);
        if (listeners == null) {
            listeners = buildMetadata(targetClass, bean);
            this.typeCache.putIfAbsent(targetClass, listeners);
        }
        listeners
                .parallelStream()
                .forEach(listener -> {
//                    String[] topics = listener.annotations.topic();
//                    for (String topic : topics) {
                    methodCache.putIfAbsent(listener.annotations.topic(), listener);
//                    }
                });
        return bean;
    }

    private List<ConsumerMethodModel> buildMetadata(Class<?> targetClass, Object bean) {
        List<ConsumerMethodModel> methods = new ArrayList<>();
        ReflectionUtils.doWithMethods(targetClass, method -> {
            Optional<Rocketmq> transactionalConsumer = findRocketmqConsumerLister(method);
            transactionalConsumer.ifPresent(
                    rocketmqConsumer
                            ->
                            methods.add(new ConsumerMethodModel(
                                            targetClass,
                                            checkProxy(method, bean),
                                            rocketmqConsumer
                                    )
                            )
            );
        }, ReflectionUtils.USER_DECLARED_METHODS);
        return methods;
    }

    private Optional<Rocketmq> findRocketmqConsumerLister(Method method) {
        return Optional.ofNullable(AnnotationUtils.findAnnotation(method, Rocketmq.class));
    }

    private Method checkProxy(Method method, Object bean) {
        if (AopUtils.isJdkDynamicProxy(bean)) {
            // 使用jdk代理处理
            try {
                method = bean.getClass().getMethod(method.getName(), method.getParameterTypes());
                Class<?>[] proxiedInterfaces = ((Advised) bean).getProxiedInterfaces();
                for (Class<?> proxiedInterface : proxiedInterfaces) {
                    try {
                        method = proxiedInterface.getMethod(
                                method.getName(),
                                method.getParameterTypes()
                        );
                        break;
                    } catch (NoSuchMethodException ignored) {
                    }
                }
            } catch (SecurityException ex) {
                ReflectionUtils.handleReflectionException(ex);
            } catch (NoSuchMethodException ex) {
                throw new IllegalStateException(String.format("方法 '%s' ,类 '%s', 找不到接口的jdk代理！"
                        , method.getName(), method.getDeclaringClass().getSimpleName()));
            }
        }
        return method;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}